package com.simplerobot.consume;
import com.simplerobot.core.AppContext;
import com.simplerobot.service.SpiderService;
import com.simplerobot.service.SpiderServiceInf;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by liangcq on 2017/3/29.
 */
public class PageConsume {
    public static Logger  log = Logger.getLogger(PageConsume.class);

    /*默认取队列中15个数据保存到数据库*/
    public final static int  PAGE_SIZE = 5;

    private AtomicInteger  count;
    /*所有的爬虫名称*/
    private List<String>  spiderNames;
    /*消费线程池*/
    private ExecutorService trueConsume;

    /*调度线程*/
    private ExecutorService dispatcher;

    /*持久化服务*/
    public SpiderServiceInf spiderService;

    private BlockingQueue<Page> pageQueue = new LinkedBlockingQueue<Page>();

    public PageConsume() {
        this.spiderNames = new ArrayList<>();
        this.trueConsume = Executors.newFixedThreadPool(4);
        this.dispatcher = Executors.newSingleThreadExecutor();
        this.spiderService = (SpiderServiceInf) AppContext.getService(SpiderService.class);
        this.count = new AtomicInteger(0);
    }


    public List<String> getSpiderNames() {
        return spiderNames;
    }
    public void setSpiderNames(List<String> spiderNames) {
        this.spiderNames = spiderNames;
    }

    public void setSpiderName(String spiderName) {
        if(!spiderNames.contains(spiderName)){
            spiderNames.add(spiderName);
        }
    }



    public void putPage(String spiderName,Page page){
        log.info(String.format("线程%s添加page到队列",Thread.currentThread().getName()));
        try {
            pageQueue.put(page);
            count.set(count.getAndIncrement());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info(String.format("线程%s添加page到队列end", Thread.currentThread().getName()));
    }

    public void eat() {
        log.info("开始消费page");
        dispatcher.execute(() -> {
            try {
                Thread.sleep(5000);
                while (true) {
                    Thread.sleep(200);
                    saveToDb();
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error(e);
            }
        });


    }

    private void saveToDb() {
        List<Page> pages=new ArrayList<>();
        int size = 0;
        Page page = null;
        while (size<PAGE_SIZE){
            try {
                page = pageQueue.take();
                count.set(count.getAndDecrement());
                log.info(String.format("临时的page大小为%d",size));
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.error(e);
            }
            if(page!=null){
                pages.add(page);
                size++;

            }
        }
        trueConsume.execute(()->{
            log.info(String.format("队列含有%s个page",count.get()));
            log.info(String.format("线程%s开始批量保存页面数据到数据库",Thread.currentThread().getName()));
            spiderService.insertBatch(Thread.currentThread().getName(),pages);
        });
    }
}
